package batch;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.ArrayList;

public class operator_filter {
    public static void main(String[] args) throws Exception {
    //    创建运行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    //    获取数据
        ArrayList<String> data = new ArrayList<String>();
        data.add("i a bing");
        data.add("i b shui");
        data.add("bing shui");
        DataSet<String> source =  env.fromCollection(data);// data 可以是个list
        //System.out.println(source.getClass().toString());
        System.out.println("*****filter****");
    //    每句话进行处理，得到返回完的结果，
        source.flatMap(new FlatMapFunction<String, String>() {
            public void flatMap(String value, Collector<String> collector) throws Exception {
                String[] words = value.split(" ");
                for(String w:words){
                    if(!w.equals("i")){
                        collector.collect(w);
                    }

                }
            }

        }).filter(new FilterFunction<String>() {
            public boolean filter(String s) throws Exception {
                // 为ture 的就返回，为false的就不返回
                //return false;
                //return true;
                return s.length() >=3 ? true:false;
            }
        }).print();


    }
}
